package com.cml.rpc.client;

import com.cml.rpc.common.RpcRequest;
import com.cml.rpc.common.RpcResponse;
import com.cml.rpc.register.ServiceRegister;
import com.cml.rpc.register.ZkServiceRegister;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;

import java.net.InetSocketAddress;

/**
 * @author cml
 * @create 2021/7/21 14:00
 */
public class NettyRpcClient implements RpcClient{
    private ServiceRegister serviceRegister;
    private static final Bootstrap bootstrap;
    private static final EventLoopGroup eventLoopGroup;
    private String host;
    private int port;
    // netty客户端初始化，只初始化一次，重复使用
    static {
        eventLoopGroup = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
                .handler(new NettyClientInitializer());
    }

    public NettyRpcClient() {
        this.serviceRegister = new ZkServiceRegister();
    }

    public NettyRpcClient(ServiceRegister serviceRegister) {
        this.serviceRegister = serviceRegister;
    }

    public NettyRpcClient(String host, int port) {
        this.host = host;
        this.port = port;
    }
    @Override
    public RpcResponse sendRequest(RpcRequest rpcRequest) {
        try {
            // 使用服务注册中心
            if (serviceRegister != null) {
                InetSocketAddress address = serviceRegister.serviceDiscover(rpcRequest.getInterfaceName());
                host = address.getHostString();
                port = address.getPort();
            }
            System.out.println("--------使用netty客户端----------");
            ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
            Channel channel = channelFuture.channel();
            // 发送数据
            channel.writeAndFlush(rpcRequest);
            channel.closeFuture().sync();

            AttributeKey<RpcResponse> key = AttributeKey.valueOf("RpcResponse");
            RpcResponse rpcResponse = channel.attr(key).get();
            return rpcResponse;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return null;
    }
}
